package net.bw.realtime.game.job;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.mongodb.source.MongoSource;
import org.apache.flink.connector.mongodb.source.enumerator.splitter.PartitionStrategy;
import org.apache.flink.connector.mongodb.source.reader.deserializer.MongoDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.bson.BsonDocument;

/*
 * @ Author：liuyawei
 * @ Date：2025-06-13
 */
public class MongoDBSourceJob {

    public static void main(String[] args) throws Exception {

        // 1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 2.创建数据源
        MongoSource<String> source = MongoSource.<String>builder()
                // 连接信息
                .setUri("mongodb://localhost:27017")
                // 数据库
                .setDatabase("db")
                // 表
                .setCollection("game_sales")
                // 筛选字段
                .setProjectedFields("_id", "game_id", "platform_id","sale_amt","sale_date","sale_id","units_sold")
                // 批次大小
                .setFetchSize(2048)
                // 批次数量
                .setLimit(10000)
                // 游标超时
                .setNoCursorTimeout(true)
                // 分区策略
                .setPartitionStrategy(PartitionStrategy.SAMPLE)
                // 分区大小
                .setPartitionSize(MemorySize.ofMebiBytes(64))
                // 分区样本数量
                .setSamplesPerPartition(10)
                // 反序列化
                .setDeserializationSchema(new MongoDeserializationSchema<String>() {
                    @Override
                    public String deserialize(BsonDocument document) {
                        // 返回json字符串
                        return document.toJson();
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                })
                .build();

        // 3.数据转换
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "MongoDB-Source")
                .print()
                .setParallelism(1);

        // 4.触发执行
        env.execute("MongoDB-Source");

    }

}
